package com.zhang.sql2;

import com.zhang.first.utils.ClickEvent;
import com.zhang.first.utils.ClickEventSource;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @title: Flink快速入门
 * @author: zhang
 * @date: 2022/3/19 16:06
 */
public class SimpleTableExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<ClickEvent> clickEventDS = env
                .addSource(new ClickEventSource())
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                . <ClickEvent>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(new SerializableTimestampAssigner<ClickEvent>() {
                                    @Override
                                    public long extractTimestamp(ClickEvent element, long recordTimestamp) {
                                        return element.ts;
                                    }
                                })
                );

        //2 创建表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 3 将DataStream转化为表
        //TODO 方式一：
        Table eventTable = tableEnv.fromDataStream(clickEventDS);
        Table resultTable = tableEnv.sqlQuery("select name,url,ts from " + eventTable);
        //TODO 方式二：
        Table eventTable2 = tableEnv.fromDataStream(clickEventDS,$("ts"),$("name").as("username"));
        //tableEnv.toChangelogStream(eventTable2).print();
        //TODO 方式三：
        tableEnv.createTemporaryView("eventTable3",clickEventDS,$("ts"),$("name").as("username"));
        Table sqlQuery = tableEnv.sqlQuery("select * from eventTable3");
        //tableEnv.toDataStream(sqlQuery).print();

        tableEnv.createTemporaryView("clickTable",eventTable);
        Table aggResult = tableEnv.sqlQuery("" +
                " select name,count(url) cnt " +
                " from clickTable " +
                " group by name");

        // TODO 表转流
        tableEnv.toDataStream(resultTable).print();
        //默认追加数据
        //tableEnv.toDataStream(aggResult).print();
        //tableEnv.toChangelogStream(aggResult).print("agg");

        env.execute();
    }
}
